package com.atguigu.gamll.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gamll.realtime.app.dwd.BaseApp;
import com.atguigu.gamll.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.text.SimpleDateFormat;

public class UniqueVisitorApp extends BaseApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.创建基本环境
        // TODO 2.设置并行度
        UniqueVisitorApp uniqueVisitorApp = new UniqueVisitorApp();
        uniqueVisitorApp.baseEntry();
    }

    @Override
    public void biz(StreamExecutionEnvironment env) {
        // TODO 3.消费kafka
        // 3.1 消费的主题及消费者组
        String topic = "dwd_page_log";
        String groupId = "unique_visitor_app_group";
        // 3.2 获取消费者对象
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        // 3.3 消费数据封装成流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaSource);

        // TODO 4.对读取的数据进行类型转换
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);

//        jsonObjDS.print(">>>");

        // TODO 5.过滤
        // 5.1 按照mid进行分组
        KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"));
        // 5.2 使用flink状态编程，完成数据过滤
        SingleOutputStreamOperator<JSONObject> filterDS = keyedDS.filter(new RichFilterFunction<JSONObject>() {

            private ValueState<String> lastVisitDateState;
            private SimpleDateFormat sdf;

            // 初始化
            @Override
            public void open(Configuration parameters) throws Exception {
                // 日期格式化
                sdf = new SimpleDateFormat("yyyyMMdd");
                // 状态初始化
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("lastVisitDateState", String.class);
                // 设置状态过期时间
                StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.days(1)).build();
                valueStateDescriptor.enableTimeToLive(stateTtlConfig);

                lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                // 获取当前数据的 lastPageId
                String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                // 判断lastPageId是否有数据，如果有表示从其它页面跳转过来的，直接过滤
                if (lastPageId != null && lastPageId.length() > 0) {
                    return false;
                }
                // 从状态中获取上次访问日期
                String lastVisitDate = lastVisitDateState.value();
                String curVisitDate = sdf.format(jsonObject.getLong("ts"));
                // 判断是否是第一次登录
                if (lastVisitDate != null && lastVisitDate.length() > 0 && curVisitDate.equals(lastVisitDate)) {
                    // 访问过，这次不在算是独立访客了，直接过滤
                    return false;
                } else {
                    // 第一次访问，将访问日期添加到状态中
                    lastVisitDateState.update(curVisitDate);
                    return true;
                }
            }
        });

        filterDS.print(">>>");

        // TODO 6.将过滤后的数据写道kafka的dwd_unique_visitor主题中
        filterDS
                .map(JSONAware::toJSONString)
                .addSink(MyKafkaUtil.getKafkaSink("dwd_unique_visitor"));
    }
}
